package com.atguigu.chapter07;

import com.atguigu.chapter05.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/6 9:12
 */
public class Flink23_ProcessFunction_TimerDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(new MapFunction<String, WaterSensor>() {
                    @Override
                    public WaterSensor map(String value) throws Exception {
                        // 切分
                        String[] line = value.split(",");
                        return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));

                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<WaterSensor>forMonotonousTimestamps()
                                .withTimestampAssigner((value, ts) -> value.getTs() * 1000L)
                );

        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());

        SingleOutputStreamOperator<String> resultDS = sensorKS
                .process(
                        new KeyedProcessFunction<String, WaterSensor, String>() {
                            @Override
                            public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                                // TODO 定时器
//                                // 1. 处理时间的定时器： 注册、删除
////                                ctx.timerService().registerProcessingTimeTimer();
////                                ctx.timerService().deleteProcessingTimeTimer();
////
////                                // 2. 事件时间的定时器： 注册、删除
////                                ctx.timerService().registerEventTimeTimer();
////                                ctx.timerService().deleteEventTimeTimer();
////
////                                // 3. 当前进度： 分为 处理时间、事件时间
////                                ctx.timerService().currentProcessingTime();
////                                ctx.timerService().currentWatermark();

                                // TODO 处理时间演示
//                                long currentTs = System.currentTimeMillis();
//                                long timerTs = currentTs + 5000L;
//                                ctx.timerService().registerProcessingTimeTimer(timerTs);
//                                System.out.println("当前时间是" + currentTs + ",注册了一个 " + timerTs + " 的定时器...");

                                // TODO 事件时间演示
                                ctx.timerService().registerEventTimeTimer(5000L);
                                System.out.println("当前时间是" + ctx.timestamp() + ",注册了一个 " + 5000L + " 的定时器...");
                            }

                            @Override
                            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                                System.out.println("现在时间是" + timestamp + "定时器触发了");
                            }
                        }
                );

        resultDS.print();


        env.execute();
    }
}

/*
    定时器的原理：
        1、注册
            eventTimeTimersQueue.add(new TimerHeapInternalTimer<>(time, (K) keyContext.getCurrentKey(), namespace));
            => 每次调用注册的方法，都会 new一个定时器对象，放进一个 队列里（会去重 => 分组内去重，不同组之间没关系）
                => 如果重复注册定时器，最终还是只有一个起作用
        2、触发
            timer.getTimestamp() <= time
                time = watermark.getTimestamp()
            => watermark >= 注册的时间 ，触发定时器 （注意 watermark生成的时候，减去了 1ms）

 */